package avicit.bdp.dcs.stream.service;

import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Collections;
import java.util.Properties;
import java.util.Random;
import java.util.Set;

public class Test {

  private static final String ZK_URL = "192.168.0.25:2181";
  private static final String KAFKA_URL = "192.168.0.25:9092";
  private static String topic = "kafka2hive"; // 定义主题

  public static void main(String[] args) {
    Properties properties = new Properties();
    // kafka地址，多个地址用逗号分割
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_URL);
    ListTopicsResult result = KafkaAdminClient.create(properties).listTopics();
    KafkaFuture<Set<String>> set = result.names();

    try {
      System.out.println("kafka topic");
      System.out.println(set.get());
    } catch (Exception e) {
      e.printStackTrace();
    }

    //        producer();
    //        consumer();
  }

  public static void producer() {
    Properties properties = new Properties();
    // kafka地址，多个地址用逗号分割
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_URL);
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

    try {
      while (true) {
        String msg = "Hello," + new Random().nextInt(100);
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, msg);
        kafkaProducer.send(record);
        System.out.println("消息发送成功:" + msg);
        Thread.sleep(3000);
      }
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      kafkaProducer.close();
    }
  }

  public static void consumer() {
    Properties properties = new Properties();
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_URL);
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "kafka2hive");

    KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
    kafkaConsumer.subscribe(Collections.singletonList(topic)); // 订阅消息

    while (true) {
      ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
      for (ConsumerRecord<String, String> record : records) {
        System.out.println(
            String.format(
                "topic:%s,offset:%d,消息:%s", record.topic(), record.offset(), record.value()));
      }
    }
  }
}
